package day06;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import po.UserEvent;

/**
 * @Description: 基于间隔的join
 * @Author ZYX
 * @Date 2022/2/15 21:41
 * @Version 1.0
 */
public class Demo04 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserEvent> orderStream = env.fromElements(
                UserEvent.of("user-1", "order", 20 * 60 * 1000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
                            @Override
                            public long extractTimestamp(UserEvent UserEvent, long l) {
                                return UserEvent.timestamp;
                            }
                        })
        );

        SingleOutputStreamOperator<UserEvent> pvStream = env.fromElements(
                UserEvent.of("user-1", "pv", 5 * 60 * 1000L),
                UserEvent.of("user-1", "pv", 10 * 60 * 1000L),
                UserEvent.of("user-1", "pv", 12 * 60 * 1000L),
                UserEvent.of("user-1", "pv", 22 * 60 * 1000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
                            @Override
                            public long extractTimestamp(UserEvent UserEvent, long l) {
                                return UserEvent.timestamp;
                            }
                        })
        );

        orderStream.keyBy(obj -> obj.userId)
                .intervalJoin(pvStream.keyBy(obj -> obj.userId))
                .between(Time.minutes(-10),Time.minutes(0))
                .process(new ProcessJoinFunction<UserEvent, UserEvent, String>() {
                    @Override
                    public void processElement(UserEvent orderEvent, UserEvent pvEvent, Context context, Collector<String> out) throws Exception {
                        out.collect(orderEvent+"=>"+pvEvent);
                    }
                }).print();


        env.execute();
    }

}
